package com.atguigu.bigdata.edu.realtime.app.dwd;

import com.atguigu.bigdata.edu.realtime.app.BaseSQLApp;
import com.atguigu.bigdata.edu.realtime.common.Constant;
import com.atguigu.bigdata.edu.realtime.util.DimUtil;
import com.atguigu.bigdata.edu.realtime.util.JdbcUtil;
import com.atguigu.bigdata.edu.realtime.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.sql.Connection;

public class Dwd_06_DwdTestExamDetail extends BaseSQLApp {
    public static void main(String[] args) {
        new Dwd_06_DwdTestExamDetail().init(3210,2,"Dwd_06_DwdTestExamDetail");
    }

    @Override
    protected void handle(StreamExecutionEnvironment env, StreamTableEnvironment tEnv) {
        readOdsDb(tEnv,"Dwd_06_DwdTestExamDetail");
        readTestPaper(tEnv);
        Table test_exam = tEnv.sqlQuery("select \n" +
                "data['id'] id , \n" +
                "cast(data['paper_id'] as bigint) paper_id , \n" +
                "data['user_id'] user_id , \n" +
                "data['score'] score , \n" +
                "cast(data['duration_sec'] as decimal(16,2)) duration_sec,  \n" +
                "unix_timestamp(`data`['create_time'])*1000 ts," +
                "pt \n" +
                "from ods_db\n" +
                "where `type` = 'insert' \n" +
                "and `table` = 'test_exam' \n" +
                "and `database` = 'edu' ");
        tEnv.createTemporaryView("test_exam",test_exam);

        Table result = tEnv.sqlQuery("select\n" +
                "te.id,\n" +
                "te.paper_id,\n" +
                "te.user_id,\n" +
                "tp.course_id ,\n" +
                "cast(te.score as int) ,\n" +
                "te.duration_sec,\n" +
                "te.ts\n" +
                "from test_exam te\n" +
                "join `test_paper` for system_time as of te.pt as tp\n" +
                "on tp.id = te.paper_id");
        tEnv.executeSql("create table dwd_Test_Exam_Detail(\n" +
                "id string ,\n" +
                "paper_id bigint,\n" +
                "user_id string,\n" +
                "course_id bigint,\n" +
                "score int,\n" +
                "duration_sec decimal(16,2),\n" +
                "ts bigint ," +
                "primary key(id) not enforced ) " + SQLUtil.getUpsertKafkaSink(Constant.TOPIC_DWD_TEST_EXAM_DETAIL));
        result.executeInsert("dwd_Test_Exam_Detail");
    }
    //我来写，别动
}
